package com.mininglamp.online

import com.mongodb.casbah.commons.MongoDBObject
import com.mongodb.casbah.{MongoClient, MongoClientURI}
import com.mongodb.spark.MongoSpark
import com.mongodb.spark.config.ReadConfig
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.bson.Document
import redis.clients.jedis.Jedis

import scala.collection.mutable._
import scala.collection.immutable


object ConnHelper extends Serializable {

  //变量懒加载
  lazy val jedis = new Jedis("master", 6379)
  lazy val mongoClient = MongoClient(MongoClientURI("mongodb://master:27017/recommender"))
}

/**
  * Project: ECommerceRecommendSystem
  * Package: com.mininglamp.online
  * Description：实时推荐模块
  *
  * Created by ZhouPeng on 2022/01/06 10:17
  **/
object OnlineRecommender {

  val MONGODB_RATING_COLLECTION = "Rating"

  //mongo中存储用户商品推荐矩阵数据集
  val USER_RECS_COLLECTION = "userRecs"
  //mongo中存储商品相似度矩阵数据集
  val PRODUCT_RECS_COLLECTION = "productRecs"
  //redis用户最近k次评分
  val MAX_USER_RATING_NUM = 20
  //商品相似矩阵商品相似列表长度
  val MAX_SIM_PRODUCTS_NUM = 20
  //实时用户推荐列表
  val STREAM_RECS_COLLECTION = "streamRecs"

  def main(args: Array[String]): Unit = {

    //基本配置
    val config = Map(
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://master:27017/recommender",
      "mongo.db" -> "recommender",
      "kafka.topic" -> "recommender"
    )

    val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("OnlineRecommender")
    val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    val ssc = new StreamingContext(sparkSession.sparkContext, Seconds(2))

    //import sparkSession.implicits._
    implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

    val readConfig = ReadConfig(Map(
      "uri" -> mongoConfig.mongoUri,
      "collection" -> PRODUCT_RECS_COLLECTION
    ))
    //加载相似度矩阵，并广播出去
    val simProductsMatrix = MongoSpark.load(sparkSession.sparkContext, readConfig).map(
      document => {
        val productId = document.getInteger("productId")
        val docs = document.get("recs").asInstanceOf[java.util.ArrayList[Document]]
        import scala.collection.JavaConversions._
        val ab = new ArrayBuffer[Recommendation]
        for (doc <- docs) {
          ab += Recommendation(doc.getInteger("productId").toInt, doc.getDouble("score"))
        }
        ProductRecs(productId.toInt, ab)
      }
    ).map(
      productsRecs => {
        (productsRecs.productId, productsRecs.recs.map(x => (x.productId, x.score)).toMap)
      }
    ) //转换为map形式，方便相似度查询
      .collectAsMap()

    //广播商品相似矩阵
    val simProductsMatrixBC = sparkSession.sparkContext.broadcast(simProductsMatrix)

    //kafka参数
    val kafkaParam = Map(
      "bootstrap.servers" -> "master:9092,slave1:9092,slave2:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "recommender",
      "auto.offset.reset" -> "latest"
    )

    //创建kafkaDstream
    val kafkaDStream = KafkaUtils.createDirectStream(ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Seq(config("kafka.topic")), kafkaParam))

    //创建评分流数据  userId|productId|score|timestamp
    val ratingDStream = kafkaDStream.map(
      msg => {
        val dataArrays = msg.value().split("\\|")
        (dataArrays(0).toInt, dataArrays(1).toInt, dataArrays(2).toDouble, dataArrays(3).toLong)
      }
    )

    //核心算法部分
    ratingDStream.foreachRDD(
      rdd => rdd.foreachPartition(
        partition => {
          partition.foreach {
            case (userId, productId, score, timestamp) => {
              println("rating data comming!>>>>>>>>>>>>>>>>>>")

              //todo 核心算法流程
              //1. 从redis中取出当前用户最近评分，保存成一个数组Array[(productId,score)]
              val userRecentlyRatings = getUserRecentlyRatings(MAX_USER_RATING_NUM,
                userId, ConnHelper.jedis)

              //2. 从相似度矩阵中获取当前商品最相似的商品列表，作为备选商品列表，保存成一个数组Array[productId]
              val candidateProducts = getTopSimProducts(MAX_SIM_PRODUCTS_NUM, productId,
                userId, simProductsMatrixBC.value)

              //3. 计算每个备选商品的推荐优先级，得到该用户的实时推荐列表，保存成一个数组 Array[(productId,score)]
              val streamRecs = computeSimProductScore(userRecentlyRatings, candidateProducts, simProductsMatrixBC.value)

              //4. 把推荐列表写到mongodb
              saveData2MongoDB(userId, streamRecs)
            }
          }
        }
      )
    )

    //启动sparkstreaming程序
    ssc.start()
    println("streaming started!!!!")
    ssc.awaitTermination()

  }

  import scala.collection.JavaConversions._

  /**
    * 从redis中获取用户最近k次评分数据
    *
    * @param num
    * @param userId
    * @param jedis
    * @return
    */
  def getUserRecentlyRatings(num: Int, userId: Int, jedis: Jedis): Array[(Int, Double)] = {

    //redis中list集合数据样例，key:"userId:userId" value:"productId:score"
    val userRecentlyRatings = jedis.lrange("userId:" + userId, 0, num)
      .map(
        item => {
          val dataArrays = item.split("\\:")
          (dataArrays(0).trim.toInt, dataArrays(1).trim.toDouble)
        }
      ).sortWith(_._2 > _._2)
      .toArray

    userRecentlyRatings
  }


  /**
    * 从商品相似度矩阵中找出相近的k个商品，并且前提过滤掉该用户已经评分过的商品
    *
    * @param num
    * @param productId
    * @param userId
    * @param simProducts
    * @return
    */
  def getTopSimProducts(num: Int,
                        productId: Int,
                        userId: Int,
                        simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]])
                       (implicit mongoConfig: MongoConfig): Array[Int] = {
    //从广播变量相似矩阵中拿到当前商品的所有相似商品列表
    val allSimProducts = simProducts(productId).toArray

    val mongoClient = ConnHelper.mongoClient
    val rating_collection = mongoClient(mongoConfig.mongodb)(MONGODB_RATING_COLLECTION)
    val ratingDataByUserId = rating_collection.find(MongoDBObject("userId" -> userId))
      .toArray
      .map(
        rating => {
          //只需要拿出productId字段
          rating.get("productId").toString.toInt
        }
      )

    //过滤当前用户评分过的商品
    allSimProducts.filter(
      item => {
        !ratingDataByUserId.contains(item._1)
      }
    )
      .sortWith(_._2 > _._2)
      .take(num)
      .map(_._1)
  }


  /**
    * 根据公式计算每个备选商品的推荐优先级
    *
    * @param userRecentlyRatings
    * @param candidateProducts
    * @param simProducts
    * @return
    */
  def computeSimProductScore(userRecentlyRatings: Array[(Int, Double)],
                             candidateProducts: Array[Int],
                             simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]): Array[(Int, Double)] = {


    //定义可变长数组，保存每个备选商品的基础得分(productId,score)
    val scores = new ArrayBuffer[(Int, Double)]()
    //定义2个Map,用于保存每个商品高分和低分个数的计数器 product -> count
    val increMap = new HashMap[Int, Int]()
    val decreMap = new HashMap[Int, Int]()

    for (candidateProduct <- candidateProducts; userRecentlyRating <- userRecentlyRatings) {
      //当前备选商品和当前已评分商品之间的相似度得分
      val simScore = getProductSimScore(candidateProduct, userRecentlyRating._1, simProducts)
      if (simScore > 0.4) {
        scores += ((candidateProduct, simScore * userRecentlyRating._2))
        if (userRecentlyRating._2 > 3) {
          increMap(candidateProduct) = increMap.getOrDefault(candidateProduct, 0) + 1
        } else {
          decreMap(candidateProduct) = decreMap.getOrDefault(candidateProduct, 0) + 1
        }
      }
    }

    //根据公式算出每个备选商品的推荐优先级
    scores.groupBy(_._1).map {
      case (productId, scoreList) => {
        (productId, scoreList.map(_._2).sum / scoreList.length
          + log(increMap.getOrDefault(productId, 1))
          - log(decreMap.getOrDefault(productId, 1)))
      }
    }.toArray
      .sortWith(_._2 > _._2)
  }

  /**
    * 获取商品之间的相似度评分
    *
    * @param productId1
    * @param productId2
    * @param simProducts
    */
  def getProductSimScore(productId1: Int, productId2: Int,
                         simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]): Double = {

    simProducts.get(productId1) match {
      case Some(sim) => sim.get(productId2) match {
        case Some(score) => score
        case None => 0.0
      }
      case None => 0.0
    }
  }

  /**
    * 自定义log函数,以N为底
    *
    * @param m
    * @return
    */
  def log(m: Int): Double = {
    val N = 10
    Math.log(m) / Math.log(N)
  }

  /**
    * 推荐列表写入mongo
    *
    * @param userId
    * @param streamRecs
    */
  def saveData2MongoDB(userId: Int, streamRecs: Array[(Int, Double)])(implicit mongoConfig: MongoConfig): Unit = {

    val stream_recs_collection = ConnHelper.mongoClient(mongoConfig.mongodb)(STREAM_RECS_COLLECTION)
    //val stream_recs_collection = mongoClient(mongoConfig.mongodb)(STREAM_RECS_COLLECTION)

    //按照userId查询更新
    stream_recs_collection.findAndRemove(MongoDBObject("userId" -> userId))
    stream_recs_collection.insert(MongoDBObject("userId" -> userId,
      "recs" -> streamRecs.map(x => MongoDBObject("productId" -> x._1, "score" -> x._2))))
  }

}


/**
  * mongodb配置样例类
  *
  * @param mongoUri mongo连接uri
  * @param mongodb  mongo连接数据库
  */
case class MongoConfig(mongoUri: String, mongodb: String)

/**
  * 标准推荐对象样例类
  *
  * @param productId
  * @param score
  */
case class Recommendation(productId: Int, score: Double)


/**
  * 商品相似度列表样例类
  *
  * @param productId
  * @param recs
  */
case class ProductRecs(productId: Int, recs: Seq[Recommendation])
